package txts

import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.sources.{BaseRelation, TableScan}
import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SQLContext}

class TextDataSourceRelation(override val sqlContext: SQLContext,path:String,userSchema: StructType) extends BaseRelation with TableScan with Logging{

  //如果传进来的schema不为空，就用传进来的schema，否则就用自定义的schema
  override def schema: StructType = {
    if(userSchema != null){
      userSchema
    }else{
      StructType(
        StructField("id",LongType,false) ::
          StructField("name",StringType,false) ::
          StructField("gender",StringType,false) ::
          StructField("salary",LongType,false) ::
          StructField("comm",LongType,false) :: Nil
      )
    }
  }

  //把数据读进来，读进来之后把它转换成 RDD[Row]
  override def buildScan(): RDD[Row] = {
    logWarning("this is buildScan....")
    //读取数据，变成为RDD
    //wholeTextFiles会把文件名读进来，可以通过map(_._2)把文件名去掉，第一位是文件名，第二位是内容
    val rdd = sqlContext.sparkContext.wholeTextFiles(path).map(_._2)
    //拿到schema
    val schemaField = schema.fields

    //rdd.collect().foreach(println)

    //rdd + schemaField 把rdd和schemaField解析出来拼起来
    val rows = rdd.map(fileContent => {
      //拿到每一行的数据
      val lines = fileContent.split("\n")
      //每一行数据按照逗号分隔，分隔之后去空格，然后转成一个seq集合
      val data = lines.map(_.split(",").map(_.trim)).toSeq

      //zipWithIndex
      val result = data.map(x => x.zipWithIndex.map {
        case (value, index) => {

          val columnName = schemaField(index).name
          //castTo里面有两个参数，第一个参数需要给个判断，如果是字段是性别，里面再进行判断再转换一下，如果不是性别就直接用这个字段
          Utils.castTo(if(columnName.equalsIgnoreCase("gender")){
            if(value == "0"){
              "man"
            }else if(value == "1"){
              "woman"
            } else{
              "unknown"
            }
          }else{
            value
          },schemaField(index).dataType)

        }
      })

      result.map(x => Row.fromSeq(x))
    })

    rows.flatMap(x => x)

  }
}
